package cn.jltx.rxjava.rx.learn;

import android.util.Log;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2017/08/09.
 */

public class RxJava2Learn1 {

    public static void createObservable() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG,"onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                 Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    public static void createObservablePreComplete() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                 Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                 Log.d(XqTag.TAG,"发射 onComplete事件");
                e.onComplete();
                 Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                 Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                 Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                 Log.d(XqTag.TAG,"onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                 Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    public static void createObservablePreErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                 Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                 Log.d(XqTag.TAG,"发射 onError");
                e.onError(new NullPointerException());
                 Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                 Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                 Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                 Log.d(XqTag.TAG,"onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                 Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    /**
     * 演示发送多个error报错
     */
    public static void createObservableMutilErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                 Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                 Log.d(XqTag.TAG,"发射 onError事件");
                e.onError(new NullPointerException());
                 Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                 Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                 Log.d(XqTag.TAG,"发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                 Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                 Log.d(XqTag.TAG,"onNext:" + value);
            }

            @Override
            public void onError(Throwable e) {
                 Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    /**
     * Dispose中断演示
     */
    public static void createObservableDispose() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                 Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                 Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                 Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                 Log.d(XqTag.TAG,"发射 D");
                e.onNext("D");
                 Log.d(XqTag.TAG,"发射 onComplete事件");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {
            private Disposable disposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
                 Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                i++;
                 Log.d(XqTag.TAG,value);
                if (i == 2) {
                     Log.d(XqTag.TAG,"disponsable---");
                    disposable.dispose();
                     Log.d(XqTag.TAG,"isDisponsable:" + disposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                 Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    public static void createFlowable() {

        // create a flowable
        Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(FlowableEmitter<String> e) throws Exception {
                e.onNext("Hello Rx2");
                e.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

        // create subscriber
        Subscriber subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                 Log.d(XqTag.TAG,"onSubscribe");
                //调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。
                // 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(String s) {
                 Log.d(XqTag.TAG,"onNext:" + s);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        };

        flowable.subscribe(subscriber);
    }


    public static void crateFlowableMerge(){
        List<Observable<String>> flowableList = new ArrayList<>();
        flowableList.add(Observable.just("任务1"));
        flowableList.add(Observable.just("任务2").doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
//                throw new Exception("任务2空");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                Log.d(XqTag.TAG,"onError"+throwable.toString());
            }
        }));
        flowableList.add(Observable.just("任务3"));
        flowableList.add(Observable.just("任务4").doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
//                throw new Exception("任务4空");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
//                Log.d(XqTag.TAG,"onError"+throwable.toString());
            }
        }));
        flowableList.add(Observable.just("任务5"));
        flowableList.add(Observable.just("任务6").doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
//                throw new Exception("任务6空");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
//                Log.d(XqTag.TAG,"onError"+throwable.toString());
            }
        }));

            Observable.zip(flowableList, new Function<Object[], List<String>>() {
                @Override
                public List<String> apply(Object[] objects) throws Exception {
                    List<String> result = new ArrayList<>();
                    if(objects != null){
                        for (Object o : objects){
                            result.add((String)o);
                        }
                    }
                    return result;
                }
            }).subscribe(new Observer<List<String>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    Log.d(XqTag.TAG,"onSubscribe:");
                }

                @Override
                public void onNext(List<String> strings) {
                    Log.d(XqTag.TAG,"onNext:" + strings.size());
                }

                @Override
                public void onError(Throwable e) {
                    Log.d(XqTag.TAG,"onError:" + e);
                }

                @Override
                public void onComplete() {
                    Log.d(XqTag.TAG,"onComplete");
                }
            });
//        Observable.mergeDelayError(flowableList)
//                .flatMap(new Function<String, ObservableSource<String>>() {
//                    @Override
//                    public ObservableSource<String> apply(String s) throws Exception {
//                        return Observable.just(s);
//                    }
//                })
//                .subscribe(new Observer<String>() {
//                    @Override
//                    public void onSubscribe(Disposable d) {
//                        Log.d(XqTag.TAG,"onSubscribe:");
//                    }
//
//                    @Override
//                    public void onNext(String s) {
//                        Log.d(XqTag.TAG,"onNext:" + s);
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        Log.d(XqTag.TAG,"onError"+e.toString());
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        Log.d(XqTag.TAG,"onComplete");
//                    }
//                });
//                .subscribe(new FlowableSubscriber<String>() {
//                    Subscription subscription;
//                    int maxCount = 2;
//                    @Override
//                    public void onSubscribe(Subscription s) {
//                        Log.d(XqTag.TAG,"onSubscribe:" + s);
//                        subscription = s;
//                        subscription.request(maxCount);
//
//                    }
//
//                    @Override
//                    public void onNext(String s) {
//                        Log.d(XqTag.TAG,"onNext:" + s);
//                        subscription.request(1);
//                    }
//
//                    @Override
//                    public void onError(Throwable t) {
//                        Log.d(XqTag.TAG,"onError"+t.toString());
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        Log.d(XqTag.TAG,"onComplete");
//                    }
//                });
    }

    public static void createFlowableJust() {
        Flowable<String> flowable = Flowable.just("Hello Rx2", "ceshi");

        //仅仅关心onNext方法。所以可以简写成下面这样
        Consumer consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                 Log.d(XqTag.TAG,"accept:" + s);
            }
        };

        // create subscriber
        Subscriber subscriber = new Subscriber<String>() {
            Subscription subscription;
            @Override
            public void onSubscribe(Subscription s) {
                 Log.d(XqTag.TAG,"onSubscribe");
                //调用request去请求资源，参数就是要请求的数量，一般如果不限制请求数量，可以写成Long.MAX_VALUE。
                // 如果你不调用request，Subscriber的onNext和onComplete方法将不会被调用
                subscription = s;
                subscription.request(1);
            }

            @Override
            public void onNext(String s) {
                 Log.d(XqTag.TAG,"onNext:" + s);
                subscription.request(1);
            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {
                 Log.d(XqTag.TAG,"onComplete");
            }
        };

        flowable.subscribe(subscriber);
    }

    /**
     * 先发射erro接着发射complete,不会报错，但上游发射onErro事件后还可以继续发射，而下游接收了onErro事件之后并不会再继续接收
     */
    public  static void createObservableErroComplete() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                Log.d(XqTag.TAG,"发射 onError事件");
                e.onError(new NullPointerException());
                Log.d(XqTag.TAG,"发射 D");
                e.onNext("D");
                Log.d(XqTag.TAG,"发射 onComplete事件");
                e.onComplete();
            }
        }).subscribe(new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG,value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    /**
     * 先发射complete接着发射erro,会报错
     */
    public  static void createObservableCompleteErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                Log.d(XqTag.TAG,"发射 onComplete事件");
                e.onComplete();
                Log.d(XqTag.TAG,"发射 D");
                e.onNext("D");
                Log.d(XqTag.TAG,"发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                Log.d(XqTag.TAG,value);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG,"onComplete");
            }
        });
    }

    public static void createObservableDisposeErro() {
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                Log.d(XqTag.TAG,"发射 A");
                e.onNext("A");
                Log.d(XqTag.TAG,"发射 B");
                e.onNext("B");
                Log.d(XqTag.TAG,"发射 C");
                e.onNext("C");
                Log.d(XqTag.TAG,"发射 D");
                e.onNext("D");
                Log.d(XqTag.TAG,"发射 onError事件");
                e.onError(new NullPointerException());
            }
        }).subscribe(new Observer<String>() {
            private Disposable disposable;
            private int i;

            @Override
            public void onSubscribe(Disposable d) {
                disposable = d;
                Log.d(XqTag.TAG,"onSubscribe");
            }

            @Override
            public void onNext(String value) {
                i++;
                Log.d(XqTag.TAG,value);
                if (i == 2) {
                    Log.d(XqTag.TAG,"disponsable---");
                    disposable.dispose();
                    Log.d(XqTag.TAG,"isDisponsable:" + disposable.isDisposed());
                }
            }

            @Override
            public void onError(Throwable e) {
                Log.d(XqTag.TAG,"onError");
            }

            @Override
            public void onComplete() {
                Log.d(XqTag.TAG,"onComplete");
            }
        });
    }
    

}
